package jp.ne.hatena.d.digitalsoul.warikan.repositories.impl

import java.util.ArrayList
import java.util.HashMap
import java.util.UUID

import scala.collection.JavaConversions._

import org.apache.log4j.Logger
import org.apache.cassandra.thrift._
import ConsistencyLevel._

import org.apache.thrift.protocol.TBinaryProtocol
import org.apache.thrift.protocol.TProtocol
import org.apache.thrift.transport.TSocket
import org.apache.thrift.transport.TTransport

import jp.ne.hatena.d.digitalsoul.warikan.models._
import jp.ne.hatena.d.digitalsoul.warikan.cassandra._
import CassandraMutation._
import CassandraRange._

class SlopeRepositoryImpl {
	
	private val logger:Logger = Logger.getLogger(classOf[PartyRepositoryImpl])
	
	def nextId:String = UUID.randomUUID.toString
	
	def add(slope:Slope) = {
		
		logger.info("slope added. slopeId:" + slope.id)
		
		val timestamp = System.currentTimeMillis
		
		// inner map
		val innerMap = new HashMap[String, java.util.List[Mutation]] // CFName, Column

		val slopeId = slope.id
		
		// slope
		val slopeRows = new ArrayList[Mutation]
		val slopeColumnMutation:Mutation = createColumnMutation("slopeName".getBytes("utf-8"), slope.name.getBytes("utf-8"), timestamp)
		slopeRows.add(slopeColumnMutation)

		innerMap.put("Slope", slopeRows)
		
		// weight
		val weightRows = new ArrayList[Mutation]
		slope.weights.foreach { it =>
		
			// mapping id
			val mappingId = UUID.randomUUID.toString
		
			// columns
			val roleName:Column = new Column("roleName".getBytes("utf-8"), it._1.getBytes("utf-8"), timestamp)
			val weight:Column = new Column("weight".getBytes("utf-8"), it._2.toString.getBytes("utf-8"), timestamp)
			
			val mapping:java.util.List[Column] = new ArrayList[Column]()
			mapping.add(roleName)
			mapping.add(weight)
						
			val rolesSuperColumnMutation:Mutation = createSuperColumnMutation(mappingId.getBytes("utf-8"), mapping)
			
			weightRows.add(rolesSuperColumnMutation)
			
		}
		innerMap.put("Weight", weightRows)
		
		// mutation map
		val mutationMap = new HashMap[String, java.util.Map[String, java.util.List[Mutation]]]() // key, innerMap
		mutationMap.put(slopeId, innerMap)
		
		// insert
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open			
			
		try {
			client.batch_mutate("warikan", mutationMap, ONE)
		} finally {
			transport.close
		}
	}
	
	def forSlopeId(id:String):Option[Slope] = {
				
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open
		
		try {
		
			// slope
			val slopeNameColumnPath:ColumnPath = new ColumnPath("Slope")
			slopeNameColumnPath.setColumn("slopeName".getBytes("utf-8"))
		
			var slopeNameColumn:ColumnOrSuperColumn = null
			try {
				slopeNameColumn = client.get("warikan", id, slopeNameColumnPath, ConsistencyLevel.ONE)
			} catch { 
				case ex:NotFoundException => return None
			}

			val slopeName:String = new String(slopeNameColumn.column.value, "utf-8")
			val slope = new Slope(id, slopeName)
			
			// weight
			val weightColumnParent:ColumnParent = new ColumnParent("Weight")
			val weightCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", id, weightColumnParent, sliceRangeAll, ONE)
			logger.info("weightCoSs:" + weightCoSs)
		
			weightCoSs.foreach { weightCoS =>
				populateWeights(slope, weightCoS)
			}
		
			return Option(slope)
		
		} finally {
			transport.close
		}
	}
		
	def forAll:List[Slope] = {
		
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open

		try {
		
			// slope
			val slopeColumnParent:ColumnParent = new ColumnParent("Slope")
			val slopeKeySlices:java.util.List[KeySlice] = client.get_range_slices("warikan", slopeColumnParent, sliceRangeAll, keyRangeAll, ONE)
			logger.info("slopeKeySlices:" + slopeKeySlices)
			if (slopeKeySlices isEmpty) return List[Slope]()
		
			// weights
			val weightColumnParent:ColumnParent = new ColumnParent("Weight")
		
			val keys = new ArrayList[String]()
			slopeKeySlices.foreach { slopeKeySlice =>
				keys.add(slopeKeySlice.key)
			}
				
			val weightCoSMap:java.util.Map[String, java.util.List[ColumnOrSuperColumn]] = client.multiget_slice("warikan", keys, weightColumnParent, sliceRangeAll, ONE)		
			logger.info("weightCoSMap:" + weightCoSMap)
						
			// slope list
			var slopeList = List[Slope]()
			slopeKeySlices.foreach { slopeKeySlice =>
				val slope:Slope = convertToSlope(slopeKeySlice)
				weightCoSMap(slope.id).foreach { weightCoS =>
					populateWeights(slope, weightCoS)
				}
				slopeList ::= slope
			}
		
			return slopeList
		
		} finally {
			transport.close
		}
	}
	
	def weightOf(slopeId:String, role:String):Int = {
		
		val transport:TTransport = new TSocket("localhost", 9160)
		val protocol:TProtocol = new TBinaryProtocol(transport)
		val client:Cassandra.Client = new Cassandra.Client(protocol)
		transport.open

		try {
			
			// weight
			val weightColumnParent:ColumnParent = new ColumnParent("Weight")
			val weightCoSs:java.util.List[ColumnOrSuperColumn] = client.get_slice("warikan", slopeId, weightColumnParent, sliceRangeAll, ONE)
			logger.info("weightCoSs:" + weightCoSs)
			
			for (weightCoS <- weightCoSs) {
				val weightMap = convertToWeightMap(weightCoS.super_column.columns)
				if (weightMap._1 equals role) return weightMap._2
			}
			
			return 0
			
		} finally {
			transport.close
		}		
	}
	
	private def convertToSlope(keySlice:KeySlice):Slope = {
		val slopeId = keySlice.key
		val slopeName = new String(keySlice.columns(0).column.value, "utf-8")
		new Slope(slopeId, slopeName)
	}

	
	def populateWeights(slope:Slope, weightCoS:ColumnOrSuperColumn) = {
		val columns = weightCoS.super_column.columns
		slope.weights += convertToWeightMap(columns)
	}
	
	def convertToWeightMap(columns:java.util.List[Column]):(String, Int) = {
		var roleName:String = null
		var weight:Int = 0
		columns.foreach { column =>
			new String(column.name, "utf-8") match {
				case "roleName" => roleName = new String(column.value, "utf-8")
				case "weight" => weight = new String(column.value, "utf-8").toInt
			}
		}
		roleName -> weight
	}
}